package org.yuanzheng.time

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.yuanzheng.source.StationLog

/**
 * @author yuanzheng
 * @date 2020/6/22-21:39
 *       需求：每隔 5 秒中统计一下最近 10 秒内每个基站中通话时间最长的一次通话发生的呼叫时间、主叫号码，被叫号码，通话时长。并且还得告诉我到底是哪个时间范围（10 秒）内的。
 */
object MaxLongCallTime2 {
  def main(args: Array[String]): Unit = {
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    import org.apache.flink.streaming.api.scala._

    //设置时间语义
    streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //streamEnv.getConfig.setAutoWatermarkInterval(100L) //周期引入watermark的设置，默认是100毫秒

    //读取数据源
    val stream: DataStream[StationLog] = streamEnv.socketTextStream("192.168.1.10", 8888)
      .map(line => {
        var split = line.split(",")
        new StationLog(split(0).trim, split(1).trim, split(2).trim, split(3).trim, split(4).trim.toLong, split(5).trim.toLong)
      })

      //引入watermark(数据乱序，延时3秒，周期性的watermark)
      //第一种：直接采用AssignerWithPeriodicWatermarks接口实现类（Flink提供的）
      /*      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[StationLog](Time.seconds(3)) {
              override def extractTimestamp(element: StationLog): Long = { // 设置eventTime
                element.callTime
              }
            })*/
      //第二种：自己定义一个AssignerWithPeriodicWatermarks接口的实现类
      .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[StationLog] {
        var maxEventTime: Long = _

        override def getCurrentWatermark: Watermark = { //周期性的生成watermark
          new Watermark(maxEventTime - 3000L)
        }

        //设置eventTime是哪个属性
        override def extractTimestamp(element: StationLog, previousElementTimestamp: Long): Long = {
          maxEventTime = maxEventTime.max(element.callTime)
          element.callTime
        }
      })

    //分组、开窗
    stream.filter(_.callType.equals("success")).keyBy(_.sid).timeWindow(Time.seconds(10), Time.seconds(5))
      .reduce(new MyReduceFunction(), new ReturnMaxTimeWindowFunction).print()

    streamEnv.execute()

  }

  class MyReduceFunction() extends ReduceFunction[StationLog] { // 增量聚合
    override def reduce(t: StationLog, t1: StationLog): StationLog = {
      if (t.duration > t1.duration) t else t1
    }
  }

  class ReturnMaxTimeWindowFunction extends WindowFunction[StationLog, String, String, TimeWindow] { // 在窗口触发时才调用一次
    override def apply(key: String, window: TimeWindow, input: Iterable[StationLog], out: Collector[String]): Unit = {
      var value = input.iterator.next()
      var sb = new StringBuilder
      sb.append("窗口范围是:").append(window.getStart).append("---").append(window.getEnd)
        .append("\n")
        .append("呼叫时间是:").append(value.callTime)
        .append("主叫号码是:").append(value.callOut)
        .append("被叫号码是:").append(value.callIn)
        .append("通话时长是:").append(value.duration)
      out.collect(sb.toString())
    }
  }

}

